package com.serial.comm.config.rabbitmq;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.serial.comm.config.redis.RedisService;
import com.serial.comm.core.constants.RabbitConst;
import com.serial.comm.core.constants.RedisConstants;
import com.serial.comm.core.exception.NoValidTelegram;
import com.serial.comm.core.exception.enums.ResponseCodeEnum;
import com.serial.comm.core.rest.fodb.FODBRestClient;
import com.serial.comm.core.telegram.Director;
import com.serial.comm.core.telegram.builder.AftnBuilder;
import com.serial.comm.core.telegram.builder.Builder;
import com.serial.comm.core.telegram.utils.ITA2Util;
import com.serial.comm.core.utils.Pair;
import com.serial.comm.core.utils.SerialUtils;
import com.serial.comm.core.utils.StringUtils;
import com.serial.comm.core.utils.TelegramUtils;
import com.serial.comm.modular.serialPort.enums.FODBEnums;
import com.serial.comm.modular.serialPort.enums.TelegramEnum;
import com.serial.comm.modular.serialPort.model.FodbMsg;
import com.serial.comm.modular.serialPort.model.Origin;
import com.serial.comm.modular.serialPort.model.Telegram;
import com.serial.comm.modular.serialPort.model.TelegramMQ;
import com.serial.comm.modular.serialPort.service.IOriginService;
import com.serial.comm.modular.serialPort.service.ITelegramService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.CharsetEncoder;
import java.util.Date;
import java.util.List;
import java.util.UUID;

/**
 * 串口业务队列 消费者
 */
@Component
public class SerialMsgReceiver {

    public static Logger log = LoggerFactory.getLogger(SerialMsgReceiver.class);

    public static CharsetEncoder ita2Encoder = new ITA2Util().newEncoder();

    @Resource
    private FODBRestClient fodbRestClient;

    @Resource
    private IOriginService originService;

    @Resource
    private ITelegramService telegramService;

    @Resource
    private SerialMsgSender serialMsgSender;

    @Resource
    private RedisService redisService;


    /**
     * 串口读取 原始电报 业务队列消费者
     *
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = RabbitConst.BUSINESS_QUEUE_BASE_READ)
    public void receiveSerialBaseReadMsg(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.info(">>>>>>业务队列收到原始电报：{}", msg);
        TelegramMQ mqMsg = JSON.parseObject(msg, TelegramMQ.class);

        boolean ack = true;
        try {
            Pair<String, List<String>> result = TelegramUtils.matchValidTelegram(mqMsg.getOriginalText());
            if (StringUtils.isNotEmpty(result.first)) {
                throw new NoValidTelegram(ResponseCodeEnum.NO_VALID_TELEGRAM);
            }
            //保存原始业务信息
            Origin origin = new Origin(msg, TelegramEnum.SUCCESS.getCode(), TelegramEnum.READ.getCode(), UUID.randomUUID().toString());
            Integer sourceId = originService.createOriginTelegramAndGetId(origin);
            //取到有效电报信息   提取所有有效电报信息
            List<String> tgList = result.second;
            int count = 0;
            for (String one : tgList) {
                count++;
                TelegramMQ mqTg = new TelegramMQ(sourceId, one);
                //将每个有点电报 逐一发往 单一电报业务队列
                serialMsgSender.sendReadUnitaryQueueMsg(JSON.toJSONString(mqTg));
            }
            log.info("{}条有效电报已处理结束", count);
        } catch (Exception e) {
            ack = false;
            log.error(ResponseCodeEnum.HANDLE_BASE_TELEGRAM_FAILURE.getMessage(), e);
        }

        if (!ack) {
            //拒绝消费消息（丢失消息） 转给死信队列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        } else {
            // 手动签收
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

    /**
     * 串口读取 单一电报 业务队列消费者
     *
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = RabbitConst.BUSINESS_QUEUE_UNITARY_READ)
    public void receiveSerialUnitaryReadMsg(Message message, Channel channel) throws IOException {
        String messageId = message.getMessageProperties().getMessageId();
        String oneTelegram = new String(message.getBody());
        log.info(">>>>>>业务队列收到单一有效电报：{}", oneTelegram);


        boolean ack = true;
        try {
            TelegramMQ mqMsg = JSON.parseObject(oneTelegram, TelegramMQ.class);

            Director director = new Director();
            Builder builder = new AftnBuilder(mqMsg.getTelegramText());
            log.info(">>>>>>即将进入解析流程");
            long start = System.currentTimeMillis();
            director.produce(builder);
            long end = System.currentTimeMillis();
            log.info(">>>>>>解析结束，耗时：{} ms", end - start);

            //入库
            Telegram telegram = builder.getTelegram();
            telegram.setSourceId(mqMsg.getSourceId());
            telegram.setDaStartTime(new Date(start));
            telegram.setDaEndTime(new Date(end));

            telegramService.insert(telegram);
            log.info(">>>>>>有效电报已入库");
            // 向 FODB 推送有效电报信息
            serialMsgSender.sendFodbReadQueueMsg(FodbMsg.convertFromTelegram(telegram));
            log.info(">>>>>>有效电报已处理结束");
        } catch (Exception e) {
            ack = false;
            redisService.cacheValue(RedisConstants.TELEGRAM_EXCEPTION_REASON + messageId, e.getMessage(), RedisConstants.EXCEPTION_EXPIRE_TIME);
            log.error(e.getMessage(), e);
        }
        if (!ack) {
            //拒绝消费消息（丢失消息） 转给死信队列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        } else {
            // 手动签收
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }

    }


    /**
     * 串口写入 业务队列消费者
     *
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = RabbitConst.BUSINESS_QUEUE_WRITE)
    @RabbitHandler
    public void receiveSerialWriteMsg(@Payload TelegramMQ telegramMQ, Message message, Channel channel) throws IOException, InterruptedException {
        String msg = telegramMQ.getTelegramText();
        log.info("=================================写入业务开始=================================");
        log.info("写入队列收到业务消息：{}", JSON.toJSONString(msg));
        boolean ack = true;
        try {
            ByteBuffer sourceBytes = ita2Encoder.encode(CharBuffer.wrap(msg.toCharArray()));
            SerialUtils.sendToPort(sourceBytes.array());

            //入库成功写入的电报
            Origin origin = new Origin(msg, TelegramEnum.SUCCESS.getCode(), TelegramEnum.WRITE.getCode(), telegramMQ.getGuid());
            originService.createOriginTelegramAndGetId(origin);
            log.info(">>>>>>写入-电报已入库");

            log.info(">>>>>>准备通知fodb写入结果");
            String result = fodbRestClient.fodbRequireApi(FODBEnums.WRITE_SUCCESS.getMessage(), FODBEnums.WRITE_SUCCESS.getCode(), origin.getGuid());
            log.info("已通知fodb写入结果，FODB写入接口返回值:{}", JSON.toJSONString(result));
        } catch (Exception e) {
            log.error(e.getMessage());
            int retry = telegramMQ.getRetry();
            if (retry >= 3) {
                //重试超过三次，手动拒绝，消息沦为死信
                ack = false;
            } else {
                log.error("写入消息处理异常，正在进行第 {} 次重试", ++retry);
                telegramMQ.setRetry(retry);
                //继续重试
                serialMsgSender.sendWriteRetryQueueMsg(telegramMQ);
            }

        }
        if (!ack) {
            //拒绝消费消息（丢失消息） 转给死信队列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        } else {
            // 手动签收
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }

    }

  /*  @RabbitListener(queues = RabbitConst.QUEUE_FODB_READ)
    @RabbitHandler
    public void recieveReadMessage(String msg, Message message, Channel channel) throws IOException {
        log.info("消费FODB消息"+msg);
        // 手动签收
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }*/

}
